package studio.raptor.extractor.segment.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventDataInfo;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

/**
 * Created by liuyong8 on 2017/7/27.
 */
@Component
@Scope("prototype")
public class BuildJsonTask implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(BuildJsonTask.class);

    private BlockingQueue<EventDataInfo> eventDataQueue;
    private BlockingQueue<BuildJsonResult> buildJsonQueue;

    private ServerConfig serverConfig;

    private ExecutorService nodeBuildingExecutor;

    private int count;

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
            CompletionService<BuildJsonResult> nodeMsgBuildingComp =
                    new ExecutorCompletionService<BuildJsonResult>(nodeBuildingExecutor);
            List<EventDataInfo> eventDataInfos = new ArrayList<EventDataInfo>();
            List<BuildJsonResult> buildJsonResults = new ArrayList<>();
            int num = eventDataQueue.drainTo(eventDataInfos,100);
            if (num <= 0) {
                count++;
                if (count % 1000 == 0){
                    log.warn("assembled events have been builded, please speed up the assembling");
                }

                LockSupport.parkUntil(System.currentTimeMillis() + 100);
            }else{
                for (int i=0;i<num;i++) {
                    EventDataInfo eventDataInfo = eventDataInfos.get(i);
                    nodeMsgBuildingComp.submit(new NodeMsgBuildingTask(eventDataInfo));
                }

                for (int i=0;i<num;i++){
                    Future<BuildJsonResult> future = nodeMsgBuildingComp.take();
                    buildJsonResults.add(future.get());
                }

                Collections.sort(buildJsonResults);

                for (BuildJsonResult buildJsonResult : buildJsonResults) {
                    buildJsonQueue.put(buildJsonResult);
                }
            }
            }catch (Exception e){

            }

        }
    }

    public void setEventDataQueue(BlockingQueue<EventDataInfo> eventDataQueue) {
        this.eventDataQueue = eventDataQueue;
    }

    public void setBuildJsonQueue(BlockingQueue<BuildJsonResult> buildJsonQueue) {
        this.buildJsonQueue = buildJsonQueue;
    }

    public void setServerConfig(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
    }

    public void setNodeBuildingExecutor(ExecutorService nodeBuildingExecutor) {
        this.nodeBuildingExecutor = nodeBuildingExecutor;
    }
}
